package com.yunlongn.async.service.impl;

import com.yunlongn.async.content.NameThreadFactory;
import com.yunlongn.async.function.RetryScheduleAction;
import com.yunlongn.async.function.RetryScheduleFail;
import com.yunlongn.async.runnable.RetryRunnable;
import com.yunlongn.async.service.RetrySchedulerService;

import java.util.concurrent.*;

/**
 * 任务失败重试工具。 按照指定的时间。定时重试任务，如果失败则回调失败函数
 * @author yunlongn
 * @since 20210907
*/
public class SimpRetryScheduleServiceImpl implements RetrySchedulerService {

    /**
     * 生产调度线程池 定时调度
     */
    private final ScheduledExecutorService freeExecutorService;
    /**
     * 消费任务线程池
     */
    private final ExecutorService taskExecutorService;
    /**
     * 消费异步线程池 只有一个
     */
    private final ExecutorService consumerExecutorService;
    /**
     * 生成->消费队列
     */
    private final BlockingQueue<RetryRunnableFuture> blockingQueue = new LinkedBlockingQueue<>();

    public SimpRetryScheduleServiceImpl(ExecutorService taskExecutorService) {
        this.taskExecutorService = taskExecutorService;

        freeExecutorService = new ScheduledThreadPoolExecutor(1,
                new NameThreadFactory("free-schedule:", true));

        consumerExecutorService = new ThreadPoolExecutor(1, 1,
                60, TimeUnit.SECONDS,
                new SynchronousQueue<>(),
                new NameThreadFactory("consumer-task:", true));

        Runtime.getRuntime().addShutdownHook(new Thread(SimpRetryScheduleServiceImpl.this::doDestroy));

        consumerExecutorService.submit(this::consumer);

    }

    /**
     * 销毁线程池
     */
    protected void doDestroy() {
        if (!freeExecutorService.isShutdown()) {
            freeExecutorService.shutdownNow();
        }
        if (!consumerExecutorService.isShutdown()) {
            consumerExecutorService.shutdownNow();
        }
        if (!taskExecutorService.isShutdown()) {
            taskExecutorService.shutdownNow();
        }
    }


    /**
     * Future 执行类。将执行结果一起存放
     * @author yunlongn
     * @since 20210907
     */
    static class RetryRunnableFuture {
        private final RetryRunnable<?> retryRunnable;
        /**
         * 执行结果
         */
        private Future<Boolean> future;

        public RetryRunnableFuture(RetryRunnable<?> retryRunnable, Future<Boolean> future) {
            this.retryRunnable = retryRunnable;
            this.future = future;
        }

        public void reset(Future<Boolean> future) {
            this.future = future;
        }
    }

    /**
     * 死循环消费
     * @author yunlongn
     * @since 20210907
     */
    public void consumer() {
        while (!Thread.interrupted()) {
            RetryRunnableFuture future = null;
            try {
                // 从BlockingQueue取出一个队首的对象，在一分钟内，队列一旦有数据可取，则立即返回队列中的数据。
                // 否则直到时间超时还没有数据可取，返回失败
                future = blockingQueue.poll(60, TimeUnit.SECONDS);
            } catch (InterruptedException ignored) {
            }
            if (future == null) {
                continue;
            }
            Boolean isSuccess;
            try {
                isSuccess = future.future.get();
            } catch (InterruptedException | ExecutionException e) {
                // 中断异常或执行异常
                isSuccess = false;
            }
            if (isSuccess) {
                continue;
            }
            if (!future.retryRunnable.hashNext()) {
                // 没有下一个调度周期 执行失败回调
                future.retryRunnable.getAction().onFail();
                continue;
            }
            final RetryRunnableFuture retryRunnableFuture = future;
            // 新的一轮调度 调度完之后再度放回队列。重新执行
            freeExecutorService.schedule(() -> {
                Future<Boolean> tryFuture = taskExecutorService
                        .submit(retryRunnableFuture.retryRunnable);
                // 不等待结果。将返回内容放入对象中。
                retryRunnableFuture.reset(tryFuture);
                // 将对象放回队列中。等待下次调用
                blockingQueue.add(retryRunnableFuture);
                // 获取下次调度时间
            }, future.retryRunnable.next(), future.retryRunnable.getTimeUnit());
        }
    }


    @Override
    public <T> void retrySubmit(String retrySubmitId, RetryScheduleAction<T> action, int... periods) {
        this.retrySubmit(retrySubmitId, action, TimeUnit.SECONDS, periods);
    }

    @Override
    public <T> void retrySubmit(String retrySubmitId, RetryScheduleAction<T> action, TimeUnit timeUnit, int... periods) {
        RetryRunnable<T> runnable = new RetryRunnable<>(action, timeUnit, periods);
        Future<Boolean> future = taskExecutorService.submit(runnable);
        RetryRunnableFuture retryRunnableFuture = new RetryRunnableFuture(runnable, future);
        blockingQueue.add(retryRunnableFuture);
    }

}
